use futures_util::StreamExt;
use tokio::io::{AsyncRead, AsyncWrite, AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio_native_tls::native_tls;
use tokio_native_tls::TlsConnector;
use tokio_tungstenite::{client_async, WebSocketStream};
use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
use tungstenite::handshake::client::Request;
use url::Url;

use crate::errors::*;


/// 建立 WebSocket 连接（支持代理），返回抽象化的 WebSocketStream
pub async fn connect_ws_stream(
    url: &str,
    proxy: Option<String>,
) -> Result<UnboundedReceiver<String>> {
    let uri = Url::parse(url)?;
    let domain = uri.domain().ok_or_else(|| BinanceError::Other("Invalid domain".into()))?;
    let port = uri.port_or_known_default().unwrap_or(443);

    // 建立底层 TCP 或代理隧道连接
    let tcp_stream = if let Some(proxy_addr) = proxy {

        let proxy_addr = if proxy_addr.starts_with("http://") {
            &proxy_addr["http://".len()..]
        } else {
            &proxy_addr
        };

        println!("[WS] Connecting via proxy: {}", proxy_addr);

        let mut stream = TcpStream::connect(proxy_addr).await?;

        let connect_req = format!(
            "CONNECT {}:{} HTTP/1.1\r\nHost: {}:{}\r\n\r\n",
            domain, port, domain, port
        );
        stream.write_all(connect_req.as_bytes()).await?;

        let mut buf = [0u8; 1024];
        let n = stream.read(&mut buf).await?;
        let resp = String::from_utf8_lossy(&buf[..n]);
        if !resp.contains("200") {
            return Err(BinanceError::Other(format!("Proxy CONNECT failed: {}", resp)));
        }

        stream
    } else {
        TcpStream::connect((domain, port)).await?
    };

    // TLS upgrade
    let native_tls = native_tls::TlsConnector::new()?;
    let tls_connector = TlsConnector::from(native_tls);
    let tls_stream = tls_connector.connect(domain, tcp_stream).await?;

    // 建立 WebSocket
    let (ws_stream, _) = client_async(url, tls_stream).await?;

    // 封装为 channel
    let (tx, rx) = unbounded_channel();
    tokio::spawn(read_loop(ws_stream, tx));

    Ok(rx)
}

/// 支持任意 stream 类型的 WS 消费循环
async fn read_loop<S>(
    mut stream: WebSocketStream<S>,
    tx: UnboundedSender<String>,
)
where
    S: AsyncRead + AsyncWrite + Unpin + Send + 'static,
{
    while let Some(msg) = stream.next().await {
        if let Ok(msg) = msg {
            if msg.is_text() {
                let _ = tx.send(msg.to_string());
            }
        }
    }
}
